package textprocessing.datasources.enwikipedia;

import com.fasterxml.jackson.databind.JsonNode;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * Created by julianzliu on 5/5/2017.
 */
public class TransformerByQuery {

    static final Logger logger = LoggerFactory.getLogger(TransformerByQuery.class);


    private TransformerEnWikiNews transformer;
    private ExtractorByQuery extractor;
    private IndexerEnWikiNews indexer;

    public TransformerByQuery(){
        this.transformer = new TransformerEnWikiNews();
        this.extractor = new ExtractorByQuery();
        this.indexer = new IndexerEnWikiNews();
    }


    private void traversalRecords(QueryBuilder query, int length, int max_size){
        Long total = this.extractor.initialSearch( query );
        Long uppper = Math.min( max_size, total );
        int offset = 0;
        while (offset < uppper){
            List<JsonNode> hits = this.extractor.getDocuments(query, offset, length);
            traversalListRecords( hits );
            offset += length;
        }
    }

    private void traversalListRecords(List<JsonNode> records){
        int size = records.size();
        for(int i = 0; i < size; i++){
            JsonNode record = records.get(i);
            //processRecord(record);
            saveRecord( record );
        }
    }

    private void saveRecord(JsonNode record){
        String id = record.get("id").asText();
        this.indexer.saveDocumentNode( record, id);
        logger.info("[saveRecord] id = {}", id);
        System.out.println(String.format("\n\n[saveRecord] id = %s", id));

    }

    private void processRecord(JsonNode record){
        String id = record.get("id").asText();
        this.transformer.processSingleRecord(record, id );
        logger.info("[processRecord] id = {}", id);
        System.out.println(String.format("\n\n[processRecord] id = %s", id));
    }


    public void transferByTitle(String title){
        MatchQueryBuilder query = QueryBuilders.matchQuery("title", title);
        traversalRecords( query, 5, 20);
    }

    public void transferByText(String text){
        MatchQueryBuilder query = QueryBuilders.matchQuery("text", text);
        traversalRecords( query, 5, 20);
    }

    public void transerById(String id){
        JsonNode node = this.extractor.getById( id );
        saveRecord( node );
        processRecord( node );
    }

    public void deleteById(String id){
        /*
        delete para context doc
         */
    }

    /*********************************************************
     *
     * @param args
     */
    public static void main(String[] args){
        TransformerByQuery transformer = new TransformerByQuery();
        MatchQueryBuilder query = QueryBuilders.matchQuery("text", "Michelle Obama");
        transformer.traversalRecords( query, 10, 20);
    }
}
